package cn.com.itg.function.source;

import cn.com.itg.mapper.ZtmmMapper;
import cn.com.itg.pojo.domain.entity.ZTMM;
import cn.com.itg.util.SpringContextHolder;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.Serializable;
import java.util.Calendar;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class ZtmmSource extends RichSourceFunction<List<ZTMM>>  implements Serializable {
    // 定时器间隔时间(ms)
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    private ZtmmMapper ztmmMapper;

    @Override
    public void open(Configuration parameters) throws Exception {

        //这里注入mapper
        //通过spring 工厂读取配置文件
        ztmmMapper = SpringContextHolder.getBean("ztmmMapper");

    }

    @Override
    public void run(SourceContext<List<ZTMM>> ctx) throws Exception {
//        while (true) {
//            List<ZTMM> ztmms = ztmmMapper.selectList(
//                    Wrappers.<ZTMM>query()
//            );
//            List<ZTMM> collect = ztmms.stream().map(s -> s.setTimestamp(Calendar.getInstance().getTime().getTime())).collect(Collectors.toList());
//            ctx.collect(collect);
//            // 每隔10分钟执行一次
//            countDownLatch.await(10, TimeUnit.MINUTES);
//        }
        List<ZTMM> ztmms = ztmmMapper.selectList(
                Wrappers.<ZTMM>query()
        );
        List<ZTMM> collect = ztmms.stream().map(s -> s.setTimestamp(Calendar.getInstance().getTime().getTime())).collect(Collectors.toList());
        ctx.collect(collect);
    }

    @Override
    public void cancel() {

    }
}